package com.leilei;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.util.Arrays;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @author lei
 * @version 1.0
 * @date 2021/3/20 18:56
 * @des flink table 操作表一样操作数据
 */
public class Flink_SQL_2_union_view {
    public static void main(String[] args) throws Exception {
        //1.准备环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        EnvironmentSettings settings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);
//        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        //2.Source
        DataStreamSource<GirlFriend> scFriendList = env.fromCollection(Arrays.asList(new GirlFriend("小丽", "A", 22),
                new GirlFriend("小丽于", "C", 24),
                new GirlFriend("小胡", "D", 23)));
        DataStreamSource<GirlFriend> bjFriendList = env.fromCollection(Arrays.asList(new GirlFriend("小A", "A", 22),
                new GirlFriend("小花", "C", 24),
                new GirlFriend("张张", "D", 23)));
        //3.注册表
        //dataStream转为table
        Table scTable = tableEnv.fromDataStream(scFriendList, $("name"), $("cup"), $("age"));
        //dataStream转为临时视图
        tableEnv.createTemporaryView("bj", bjFriendList,  $("name"), $("cup"), $("age"));

        //4.执行查询语句
        // union 视图与表
        Table resultTable = tableEnv.sqlQuery("SELECT * FROM " + scTable + " WHERE age > 22" +
                " UNION ALL" +
                " SELECT * FROM bj WHERE name like '%小%'");

        //5.输出结果
        DataStream<GirlFriend> resultDS = tableEnv.toAppendStream(resultTable, GirlFriend.class);
        resultDS.print();

        env.execute();
    }
}